package com.study.mr.reducejoin;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

public class ReduceJoinReducer extends Reducer<Text,Orderpd,Orderpd,NullWritable> {
    private ArrayList<Orderpd> orderpdList=new ArrayList<Orderpd>();
    private Orderpd pd=new Orderpd();

    /**
     * 核心处理逻辑。接收Map整合好的数据，进行最终join操作
     * @param key
     * @param values
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<Orderpd> values, Context context) throws IOException, InterruptedException {
        //遍历当前相同key的一组values.分离数据
        for (Orderpd orderpd : values) {
            //判断当前数据来源
            if(orderpd.getTitle().contains("order")){
                try {
                    //将当前数据来源order.txt问件管理到一个集合中
                    Orderpd thisOrderpd = new Orderpd();
                    //将当前传入Orderpd复制到thisOrderpd
                    BeanUtils.copyProperties(thisOrderpd,orderpd);
                    orderpdList.add(thisOrderpd);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }

            }else {
                //将当前数据来源pd.txt文件
                //将当前传入Orderpd复制到thisOrderpd
                try {
                    BeanUtils.copyProperties(pd,orderpd);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
            }
        }

        //Join操作
        for (Orderpd op : orderpdList) {
            op.setPname(pd.getPname());
            //将结果数据写出
            context.write(op,NullWritable.get());
        }

        //清空orderlist
        orderpdList.clear();

    }
}
